#ifndef __NET_TOOLS_H__
#define __NET_TOOLS_H__

#include <stdio.h>
#include <stdio.h>
#include <unistd.h>
#include <assert.h>
#include <errno.h>
#include <signal.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>

#include <string>
#include <memory>
#include <functional>
#include <thread>

#ifndef USE_THREAD
#include "co_api.h"
#endif

#include "common/common_utils.h"

using namespace std;

const int MAX_BUFF_SIZE = 1024;

#ifdef USE_THREAD
#define PRINTF(fmt, ...) \
{ \
    printf("[time:%s, tid:%d] " fmt "\n", \
		CommonUtils::date_ms().c_str(), \
		CommonUtils::gettid(), \
		##__VA_ARGS__ \
	); \
}
#else
#define PRINTF(fmt, ...) \
{ \
    printf("[time:%s, tid:%d, cid:%d] " fmt "\n", \
		CommonUtils::date_ms().c_str(), \
		CommonUtils::gettid(), \
		CoApi::getcid(), \
		##__VA_ARGS__ \
	); \
}
#endif

struct CoClientCb
{
	function<void(int, bool)> connect_cb;
	function<void(int, const string&, bool)> send_cb;
	function<void(int, const string&, bool)> recv_cb;
	function<void(int)> close_cb;
};

struct CoServerCb
{
	function<void(int, const string&, bool, long)> send_cb;
	function<void(int, const string&, bool, long)> recv_cb;
	function<void(int)> close_cb;
};


class NetTools
{
public:
#ifdef USE_THREAD
    static void create_task(function<void()> f)
    {
        thread(f).detach();
    }
#else
    static void create_task(function<void()> f)
    {
        CoApi::create(f); 
    }
#endif

    static void echo_sender(const string& host, int port, int loop, CoClientCb cb)
    {
    	int s = socket(AF_INET, SOCK_STREAM, 0);
    	assert(s >= 0);

    	struct sockaddr_in caddr;
    	caddr.sin_family = AF_INET;
    	caddr.sin_port   = htons(port);
    	caddr.sin_addr.s_addr = inet_addr(host.c_str());

        auto beg_us = CommonUtils::now_us();
    	int ret = connect(s, (struct sockaddr*)&caddr, sizeof(caddr));
		auto end_us = CommonUtils::now_us();

		if (cb.connect_cb) {
			cb.connect_cb(s, ret == 0 ? true : false);
		}
		if (ret != 0) {
			return ;
		}

    	int i = 0;
    	while (loop < 0 || i < loop) {

			string content = CommonUtils::format_string("hello %d", i);

			ret = send_packet(s, content);
			if (cb.send_cb) {
				cb.send_cb(s, content, ret >= 0 ? true : false);
			}
			if (ret == -1) {
                PRINTF("send_packet fail, content:%s", content.c_str());
				break ;
			}

			ret = recv_packet(s, content);
			if (ret == 0) {
				if (cb.close_cb) {
					cb.close_cb(s);
				}
				break ;
			} else {
				if (cb.recv_cb) {
					cb.recv_cb(s, content, ret > 0 ? true : false);
				}
				if (ret == -1) {
                    PRINTF("recv_packet fail");
					break ;
				}
			}
	    	i++;
    	}
    	close(s);
    }

    static void echo_receiver(int fd, CoServerCb cb)
    {
        while (1) {
        
			string content;
			{
				auto beg_us = CommonUtils::now_us();
				int ret = recv_packet(fd, content);
				auto end_us = CommonUtils::now_us();
				if (ret == 0) {
					if (cb.close_cb) {
						cb.close_cb(fd);
					}
            	    break ;
				} else {
					if (cb.recv_cb) {
						cb.recv_cb(
							fd, 
							content, 
							ret > 0 ? true : false, 
							ret > 0 ? end_us - beg_us : 0
						);
					}
					if (ret == -1) {
						break ;
					}
				}
			}

			{
				auto beg_us = CommonUtils::now_us();
				int ret = send_packet(fd, content);
				auto end_us = CommonUtils::now_us();
				if (cb.send_cb) {
					cb.send_cb(
						fd, 
						content, 
						ret >= 0 ? true : false,
						ret > 0 ? end_us - beg_us : 0
					);
				}
				if (ret == -1) {
            	    PRINTF(
						"---------- %s|send fail, fd:%d, errno:%d, error:%s", 
						__FUNCTION__, 
						fd, 
						errno, 
						strerror(errno)
					);
					break ;
				}
			}
        }
        close(fd);
    }

    static void normal_sender(const string& host, int port, int loop, CoClientCb cb)
    {
    	int s = socket(AF_INET, SOCK_STREAM, 0);
    	assert(s >= 0);

    	struct sockaddr_in caddr;
    	caddr.sin_family = AF_INET;
    	caddr.sin_port   = htons(port);
    	caddr.sin_addr.s_addr = inet_addr(host.c_str());

        auto beg_ms = CommonUtils::now_ms();
    	int ret = connect(s, (struct sockaddr*)&caddr, sizeof(caddr));
		auto end_ms = CommonUtils::now_ms();

		if (cb.connect_cb) {
			cb.connect_cb(s, ret == 0 ? true : false);
		}
		if (ret != 0) {
			return ;
		}

		auto cost_ms = end_ms - beg_ms;
		if (cost_ms >= 100) {
			PRINTF("WARN|connect overtime cost:%ldms", end_ms - beg_ms);
			exit(-1);
		}

    	int i = 0;
    	while (loop < 0 || i < loop) {

			string content = CommonUtils::format_string("hello %d", i);

			ret = send_packet(s, content);
			if (cb.send_cb) {
				cb.send_cb(s, content, ret >= 0 ? true : false);
			}
			if (ret == -1) {
				break ;
			}
	    	i++;
    	}
    	close(s);
    }

    static void normal_receiver(int fd, CoServerCb cb)
    {
        while (1) {
        
			string content;
			auto beg_us = CommonUtils::now_us();
			int ret = recv_packet(fd, content);
			auto end_us = CommonUtils::now_us();
			if (ret == 0) {
				if (cb.close_cb) {
					cb.close_cb(fd);
				}
                break ;
			} else {
				if (cb.recv_cb) {
					cb.recv_cb(
						fd, 
						content, 
						ret > 0 ? true : false,
						ret > 0 ? end_us - beg_us : 0
					);
				}
				if (ret == -1) {
                    PRINTF("---------- %s|recv fail, fd:%d, errno:%d", __FUNCTION__, fd, errno);
					break ;
				}
			}
        }
        close(fd);
    }

	static void ignore_sig_pipe()
	{
		signal(SIGPIPE, SIG_IGN);
#ifdef USE_THREAD
        PRINTF("VERSION:USE_THREAD");
#else
        PRINTF("VERSION:USE_COROUTINE");
#endif
	}

	static int send_packet(int fd, const string& content)
    {
        string buffer;

        int size = htonl(content.length());

        buffer.append((char*)&size, sizeof(size));
        buffer.append(content.c_str(), content.length());

        int send_size = 0;
        int buff_size  = (int)buffer.size();
        while (send_size < buff_size) {
            int ret = send(fd, buffer.c_str() + send_size, buff_size - send_size, 0);
            if (ret == -1) {
                return -1;
            } else {
                send_size += ret;
            }
        }
        return size;
    }

    static int recv_packet(int fd, string& content)
	{
		char buf[1024];
		int ret = recv(fd, buf, sizeof(buf), 0);
		if (ret == -1) {
			return -1;
		} else if (ret == 0) {
			return 0;
		}

		int size = ntohl(*((int*)buf));
		if (ret != size + sizeof(int)) {
			PRINTF("recv packet error, fd:%d, recv_size:%d, need_size:%d", fd, ret, size + sizeof(int));
			return -1;
		}

		content.clear();
		content.append(buf + sizeof(int), size);

		return size;
	}
};

#endif

